/*
 * Copyright (c) GoKeep by gokeep.org. 2022-2023. All rights reserved
 */
package org.gokeep.communication.engine;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import jakarta.annotation.Resource;
import org.gokeep.common.config.properties.CommunicationProperties;
import org.gokeep.communication.proto.MessageProto;
import org.smartboot.socket.Protocol;
import org.smartboot.socket.extension.plugins.HeartPlugin;
import org.smartboot.socket.extension.processor.AbstractMessageProcessor;
import org.smartboot.socket.extension.protocol.FixedLengthBytesProtocol;
import org.smartboot.socket.transport.AioQuickServer;
import org.smartboot.socket.transport.AioSession;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Optional;

import static org.gokeep.common.config.GlobalConfig.DEFAULT_SOCKET_THREAD_NUM;

/**
 * 通信引擎模块
 */
@Slf4j
@Component
public class CommunicationEngine {

    @Resource
    private HeartPlugin<MessageProto.Message> heartPlugin;

    @Resource
    private CommunicationProperties communicationProperties;

    @Resource
    private AbstractMessageProcessor<MessageProto.Message> messageProcessor;

    /**
     * 通信引擎启动方法
     */
    public void start() {
        AioQuickServer server = new AioQuickServer(communicationProperties.getHost(), communicationProperties.getPort(), protocol(), messageProcessor);
        // 如果启动心跳消息，单位时间内部客户端不返回心跳ACK响应，则会主动断开连接
        if (communicationProperties.getHeart().isEnable()) {
            messageProcessor.addPlugin(heartPlugin);
        }
        try {
            // 关闭banner输出
            server.setBannerEnabled(false);
            // 启用低内存模式
            server.setLowMemory(true);
            // 设置读ReadBufferSize
            server.setReadBufferSize(communicationProperties.getReadBufferSize());
            // 设置通信服务线程数，如果没有设置，线程数为机器处理器数+1
            int threadNum = Optional.ofNullable(communicationProperties.getThreadNum()).orElse(DEFAULT_SOCKET_THREAD_NUM);
            server.setThreadNum(threadNum);
            // 启动服务
            server.start();
            log.info("Communication thread num is: {}, buffer-size: {}", threadNum, communicationProperties.getReadBufferSize());
            log.info("SmartSocket aio started on port {}", communicationProperties.getPort());
        } catch (IOException e) {
            log.error("tcp server start fail", e);
        }
    }


    /**
     * 定义消息协议
     * 把ByteArray转换为Message对象
     * @return
     */
    private Protocol<MessageProto.Message> protocol() {
        return new FixedLengthBytesProtocol<>() {
            @Override
            @SneakyThrows
            protected MessageProto.Message decode(byte[] bytes, AioSession aioSession) {
                return MessageProto.Message.parseFrom(bytes);
            }
        };
    }
}
